/*
 * Copyright (C) 2011 The Android Open Source Project
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.jy.appmarket.newnetwork;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

import android.os.Handler;
import android.os.Looper;

public class RequestQueue {

	/** Callback interface for completed requests. */
	public interface RequestFinishedListener<T> {
		/** Called when a request has finished processing. */
        void onRequestFinished(Request<T> request);
	}

	/**
	 * Used for generating monotonically-increasing sequence numbers for
	 * requests.
	 */
	private AtomicInteger mSequenceGenerator = new AtomicInteger();

	private final Set<Request<?>> mCurrentRequests = new HashSet<Request<?>>();

	private final PriorityBlockingQueue<Request<?>> mNetworkQueue = new PriorityBlockingQueue<Request<?>>();

	private static final int DEFAULT_NETWORK_THREAD_POOL_SIZE = 1;
	private final Cache mCache;

	private final Network mNetwork;

	private final ResponseDelivery mDelivery;

	private WorkDispatcher[] mDispatchers;

	@SuppressWarnings("rawtypes")
	private List<RequestFinishedListener> mFinishedListeners = new ArrayList<RequestFinishedListener>();

	public RequestQueue(Cache cache, Network network, int threadPoolSize, ResponseDelivery delivery) {
		mCache = cache;
		mNetwork = network;
		mDispatchers = new WorkDispatcher[threadPoolSize];
		mDelivery = delivery;
	}

	public RequestQueue(Cache cache, Network network, int threadPoolSize) {
		this(cache, network, threadPoolSize, new ExecutorDelivery(new Handler(Looper.getMainLooper())));
	}

	public RequestQueue(Cache cache, Network network) {
		this(cache, network, DEFAULT_NETWORK_THREAD_POOL_SIZE);
	}

	public void start() {
		stop();
		for (int i = 0; i < mDispatchers.length; i++) {
			WorkDispatcher networkDispatcher = new WorkDispatcher(mNetworkQueue, mNetwork, mCache, mDelivery);
			mDispatchers[i] = networkDispatcher;
			networkDispatcher.start();
		}
	}

	public void stop() {
		for (int i = 0; i < mDispatchers.length; i++) {
			if (mDispatchers[i] != null) {
				mDispatchers[i].quit();
			}
		}
	}

	public int getSequenceNumber() {
		return mSequenceGenerator.incrementAndGet();
	}

	public Cache getCache() {
		return mCache;
	}

	public interface RequestFilter {
		boolean apply(Request<?> request);
	}

	public void cancelAll(RequestFilter filter) {
		synchronized (mCurrentRequests) {
			for (Request<?> request : mCurrentRequests) {
				if (filter.apply(request)) {
					request.cancel();
				}
			}
		}
	}

	public void cancelAll(final Object tag) {
		if (tag == null) {
			return;
		}

		cancelAll(new RequestFilter() {
			@Override
			public boolean apply(Request<?> request) {
				return request.getTag() == tag;
			}
		});
	}

	public <T> Request<T> add(Request<T> request) {
		request.setRequestQueue(this);
		synchronized (mCurrentRequests) {
			mCurrentRequests.add(request);
		}

		// Process requests in the order they are added.
		request.setSequence(getSequenceNumber());
		request.addMarker("add-to-queue");

		mNetworkQueue.add(request);
		return request;
	}

	@SuppressWarnings("unchecked")
	<T> void finish(Request<T> request) {
		// Remove from the set of requests currently being processed.
		synchronized (mCurrentRequests) {
			mCurrentRequests.remove(request);
		}
		synchronized (mFinishedListeners) {
			for (RequestFinishedListener<T> listener : mFinishedListeners) {
				listener.onRequestFinished(request);
			}
		}
	}

	public <T> void addRequestFinishedListener(RequestFinishedListener<T> listener) {
		synchronized (mFinishedListeners) {
			mFinishedListeners.add(listener);
		}
	}

	public <T> void removeRequestFinishedListener(RequestFinishedListener<T> listener) {
		synchronized (mFinishedListeners) {
			mFinishedListeners.remove(listener);
		}
	}
}
